/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.consumer.store;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.impl.FindBrokerResult;
import com.alibaba.rocketmq.client.impl.factory.MQClientInstance;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

/**
 * Remote storage implementation
 *
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-24
 */
public class RemoteBrokerOffsetStore implements OffsetStore {
	private final static Logger log = ClientLogger.getLog();
	private final MQClientInstance mQClientFactory;
	private final String groupName;
	private final AtomicLong storeTimesTotal = new AtomicLong(0);
	private ConcurrentHashMap<MessageQueue, AtomicLong> offsetTable = new ConcurrentHashMap<MessageQueue, AtomicLong>();

	public RemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName) {
		this.mQClientFactory = mQClientFactory;
		this.groupName = groupName;
	}

	@Override
	public void load() {
	}

	// 更新所有broker下topic为TopicTest和%RETRY%ConsumerTEST_GROUP的队列
	@Override
	public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
		if (mq != null) {
			// 以MessageQueue对象为key值从offsetTable:ConcurrentHashMap<MessageQueue, AtomicLong>变量中获取values值
			AtomicLong offsetOld = this.offsetTable.get(mq);
			if (null == offsetOld) {
				// 若该values值为空，则将MessageQueue对象以及offset值（在3.1步中获取的PullResult.nextBeginOffset值）存入offsetTable变量中
				offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));
			}

			if (null != offsetOld) {
				// 若不为空，则比较已经存在的值，若大于已存在的值才更新
				if (increaseOnly) {
					MixAll.compareAndIncreaseOnly(offsetOld, offset);
				} else {
					offsetOld.set(offset);
				}
			}
		}
	}

	@Override
	public long readOffset(final MessageQueue mq, final ReadOffsetType type) {
		if (mq != null) {
			switch (type) {
				case MEMORY_FIRST_THEN_STORE :
				case READ_FROM_MEMORY : {
					// 以入参MessageQueue对象从RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap <MessageQueue,AtomicLong>变量中获取消费进度偏移量；
					AtomicLong offset = this.offsetTable.get(mq);
					if (offset != null) {
						// 若该偏移量不为null则返回该值，
						return offset.get();
					} else if (ReadOffsetType.READ_FROM_MEMORY == type) {
						// 否则返回-1；
						return -1;
					}
				}
				case READ_FROM_STORE : {
					try {
						// A) 以MessageQueue对象的brokername从MQClientInstance.brokerAddrTable中获取Broker的地址
						// B) 向Broker发送QUERY_CONSUMER_OFFSET请求码，获取消费进度Offset
						long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);
						AtomicLong offset = new AtomicLong(brokerOffset);

						// C)用上一步从Broker获取的offset更新本地内存的消费进度列表数据RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap<MessageQueue, AtomicLong>变量值；
						this.updateOffset(mq, offset.get(), false);

						// D）返回该offset值；
						return brokerOffset;
					}
					// No offset in broker
					catch (MQBrokerException e) {
						return -1;
					}
					// Other exceptions
					catch (Exception e) {
						log.warn("fetchConsumeOffsetFromBroker exception, " + mq, e);
						return -2;
					}
				}
				default :
					break;
			}
		}

		return -1;
	}

	@Override
	public void persistAll(Set<MessageQueue> mqs) {
		if (null == mqs || mqs.isEmpty())
			return;

		final HashSet<MessageQueue> unusedMQ = new HashSet<MessageQueue>();
		long times = this.storeTimesTotal.getAndIncrement();

		if (mqs != null && !mqs.isEmpty()) {
			for (MessageQueue mq : this.offsetTable.keySet()) { // offsetTable======>ConcurrentHashMap<MessageQueue, AtomicLong>
				AtomicLong offset = this.offsetTable.get(mq);
				if (offset != null) {
					if (mqs.contains(mq)) {
						try {
							// 向Broker发送UPDATE_CONSUMER_OFFSET请求码的消费进度信息；
							this.updateConsumeOffsetToBroker(mq, offset.get());
							if ((times % 12) == 0) {
								log.info("Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}", //
										this.groupName, //
										this.mQClientFactory.getClientId(), //
										mq, //
										offset.get());
							}
						} catch (Exception e) {
							log.error("updateConsumeOffsetToBroker exception, " + mq.toString(), e);
						}
					} else {
						unusedMQ.add(mq);
					}
				}
			}
		}

		if (!unusedMQ.isEmpty()) {
			for (MessageQueue mq : unusedMQ) {
				this.offsetTable.remove(mq);
				log.info("remove unused mq, {}, {}", mq, this.groupName);
			}
		}
	}

	@Override
	public void persist(MessageQueue mq) {
		// 以入参MessageQueue对象为key值从RemoteBrokerOffsetStore.offsetTable: ConcurrentHashMap<MessageQueue, AtomicLong>变量中获取偏移量offset值，
		AtomicLong offset = this.offsetTable.get(mq);
		if (offset != null) {
			try {
				// 向Broker发送UPDATE_CONSUMER_OFFSET请求码，向Broker进行消费进度信息的同步；
				this.updateConsumeOffsetToBroker(mq, offset.get());
				log.debug("updateConsumeOffsetToBroker {} {}", mq, offset.get());
			} catch (Exception e) {
				log.error("updateConsumeOffsetToBroker exception, " + mq.toString(), e);
			}
		}
	}

	/**
	 * Update the Consumer Offset, once the Master is off, updated to Slave,
	 * here need to be optimized.
	 */
	private void updateConsumeOffsetToBroker(MessageQueue mq, long offset) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
		FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
		if (null == findBrokerResult) {
			// TODO Here may be heavily overhead for Name Server,need tuning
			this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
			findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
		}

		if (findBrokerResult != null) {
			UpdateConsumerOffsetRequestHeader requestHeader = new UpdateConsumerOffsetRequestHeader();
			requestHeader.setTopic(mq.getTopic());
			requestHeader.setConsumerGroup(this.groupName);
			requestHeader.setQueueId(mq.getQueueId());
			requestHeader.setCommitOffset(offset);

			this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffsetOneway(findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
		} else {
			throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
		}
	}

	private long fetchConsumeOffsetFromBroker(MessageQueue mq) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
		// A）以MessageQueue对象的brokername从MQClientInstance.brokerAddrTable中获取Broker的地址
		FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
		if (null == findBrokerResult) {
			// TODO Here may be heavily overhead for Name Server,need tuning
			// 若没有获取到则立即调用updateTopicRouteInfoFromNameServer方法然后再次获取；
			this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
			findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
		}

		if (findBrokerResult != null) {
			QueryConsumerOffsetRequestHeader requestHeader = new QueryConsumerOffsetRequestHeader();
			requestHeader.setTopic(mq.getTopic());
			requestHeader.setConsumerGroup(this.groupName);
			requestHeader.setQueueId(mq.getQueueId());
			// B) 向Broker发送QUERY_CONSUMER_OFFSET请求码，获取消费进度Offset
			return this.mQClientFactory.getMQClientAPIImpl().queryConsumerOffset(findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
		} else {
			throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
		}
	}

	public void removeOffset(MessageQueue mq) {
		if (mq != null) {
			this.offsetTable.remove(mq);
			log.info("remove unnecessary messageQueue offset. mq={}, offsetTableSize={}", mq, offsetTable.size());
		}
	}

	@Override
	public Map<MessageQueue, Long> cloneOffsetTable(String topic) {
		Map<MessageQueue, Long> cloneOffsetTable = new HashMap<MessageQueue, Long>();
		Iterator<MessageQueue> iterator = this.offsetTable.keySet().iterator();
		while (iterator.hasNext()) {
			MessageQueue mq = iterator.next();
			if (!UtilAll.isBlank(topic) && !topic.equals(mq.getTopic())) {
				continue;
			}
			cloneOffsetTable.put(mq, this.offsetTable.get(mq).get());
		}
		return cloneOffsetTable;
	}
}
